package cn.item.buckle.business;

import org.apache.spark.util.AccumulatorV2;

import java.util.HashMap;
import java.util.Map;

public class MonitorConversionAccumulator extends AccumulatorV2<Map<String, Long>, Map<String, Long>> {
	private final Map<String, Long> monitorCars = new HashMap<>();
	@Override
	public boolean isZero() {
		return monitorCars.isEmpty();
	}

	@Override
	public AccumulatorV2<Map<String, Long>, Map<String, Long>> copy() {
		return new MonitorConversionAccumulator();
	}

	@Override
	public void reset() {
		monitorCars.clear();
	}

	@Override
	public void add(Map<String, Long> stringLongMap) {
		stringLongMap.forEach((k, v) -> {
			monitorCars.merge(k, v, Long::sum);
		});
	}

	@Override
	public void merge(AccumulatorV2<Map<String, Long>, Map<String, Long>> accumulatorV2) {
		add(accumulatorV2.value());
	}

	@Override
	public Map<String, Long> value() {
		return monitorCars;
	}
}
